/**
 *   @file ThreadPool.cpp
 */
#include "ThreadPool.h"
#include "SyncLock.h"
#include "Semaphore.h"
#include "Event.h"
#include "SyncWrap.h"
#include "Thread.h"
#include "../exceptions/BaseException.h"
#include "../util/AtomicOperations.h"
#include "../os/debug/Debug.h"
#include "../common/CppMemPool.h"


using namespace cppflib::collections;
using namespace cppflib::util;
using namespace cppflib::exceptions;
using namespace cppflib::os;

namespace cppflib
{

namespace threading
{

PRIVATE int __ThreadPoolFunc(ptr_t p);

/**
 *  Contains info about a member of a thread pool
 */
class ThreadMemberInfo : public BaseObject
{
private:
   Thread *pThread;
   Event *pDoJobEvent;
   ThreadPool::UserFunc *pUserFunc;
   ptrany_t pUserFuncArg;
   BaseFuncPtr1T<void, ThreadMemberInfo*> *pReturnToIdleFunc;
   volatile u32_t runCount;  ///< 1: run, 0: stop
   
private:
   ThreadMemberInfo() { }
   
public:
   explicit ThreadMemberInfo(BaseFuncPtr1T<void, ThreadMemberInfo*> *pReturnToIdleFunc)
   {
      pThread = new Thread((ThreadEntryPoint(&__ThreadPoolFunc)));
      cppmempool::Retain(pThread);
      pDoJobEvent = new Event();
      cppmempool::Retain(pDoJobEvent);
      pUserFunc = NULL;
      pUserFuncArg = NULL;
      this->pReturnToIdleFunc = pReturnToIdleFunc;
      cppmempool::Retain(this->pReturnToIdleFunc);
      runCount = 1;
   }

   virtual ~ThreadMemberInfo()
   {
      cppmempool::Release(pThread);
      cppmempool::Release(pDoJobEvent);
      cppmempool::Release(pUserFunc);
      cppmempool::Release(pReturnToIdleFunc);
   }

   void StartThread()
   {
      pThread->Start(reinterpret_cast<ptr_t>(this));
   }

   /**
    *  @return true -- ready to run
    *  @return false -- signal to quit
    */
   bool WaitToRunJob()
   {
      pDoJobEvent->Wait();
      if (AtomicRead(&runCount) == 0)
         return false;
      return true;
   }

   void DoJob()
   {
      try {
         (*pUserFunc)(pUserFuncArg);  // execute user provided function
      }
      catch(BaseException &e) {
         debug::Print(debug::DL_ERROR, _S("Thread function throw an exception: err = %s, errno = %d\n"), 
                                       e.GetMsg(), e.GetErrCode());
      }
   }

   void ReturnToIdle()
   {
      (*pReturnToIdleFunc)(this);
   }

   void StopThread()
   {
      AtomicDecrementFetch(&runCount);
      pDoJobEvent->Set();
      pThread->Join();
   }

   void ExecuteJob(ThreadPool::UserFunc *pUserFunc, ptrany_t pUserFuncArg)
   {
      cppmempool::Retain(pUserFunc);
      cppmempool::Release(this->pUserFunc);
      this->pUserFunc = pUserFunc;
      this->pUserFuncArg = pUserFuncArg;
      this->pDoJobEvent->Set();
   }

   friend class ThreadPool;
};

/**
 *  Thread function for each ThreadMemberInfo
 */
PRIVATE int __ThreadPoolFunc(ptr_t p)
{
   ThreadMemberInfo *pMemberInfo = reinterpret_cast<ThreadMemberInfo*>(p);

   while(pMemberInfo->WaitToRunJob()) {
      pMemberInfo->DoJob();
      pMemberInfo->ReturnToIdle();
   }

   return 0;
}


/********************** ThreadPool **************************/

ThreadPool::ThreadPool(void)
{
}

ThreadPool::ThreadPool(u32_t numThreads)
{
   this->numThreads = numThreads;
   this->pListLock = new SyncLock();
   cppmempool::Retain(this->pListLock);
   this->pIdleSem = new Semaphore(numThreads);
   cppmempool::Retain(this->pIdleSem);
   this->pBusyList = new ArrayCppObjT<ThreadMemberInfo>(static_cast<u32_t>(numThreads));
   cppmempool::Retain(this->pBusyList);
   this->pIdleList = new ArrayCppObjT<ThreadMemberInfo>(static_cast<u32_t>(numThreads));
   numIdleThreads = numThreads;
   numBusyThreads = 0;
   cppmempool::Retain(this->pIdleList);
}

ThreadPool::~ThreadPool(void)
{
   cppmempool::Release(pIdleList);
   cppmempool::Release(pBusyList);
   cppmempool::Release(pIdleSem);
   cppmempool::Release(pListLock);
}

/**
 *  Get an idle thread member. ThreadMemberInfo is being detached from idle list and attached to busy list
 */
ThreadMemberInfo *ThreadPool::GetIdleMember()
{
   pIdleSem->Wait(); // wait for an available idle thread
   
   SyncWrap<SyncLock> wrap(pListLock);

   ThreadMemberInfo * pMemberInfo = NULL;
   int idleIndex = 0;
   for (int i = 0; i < static_cast<int>(pIdleList->GetSize()); ++i) {
      pMemberInfo = pIdleList->Get(i);   
      if (pMemberInfo) {                  // pick an idle member
         idleIndex = i;
         break;
      }
   }

   for (int i = 0; i < static_cast<int>(pBusyList->GetSize()); ++i) {
      if (pBusyList->Get(i) == NULL) {    // add to empty slot
         pBusyList->Set(i, pMemberInfo);
         break;
      }
   }

   pIdleList->Remove(idleIndex);          // remove reference in idle list
   AtomicDecrementFetch(&numIdleThreads); // update
   AtomicIncrementFetch(&numBusyThreads); // update

   return pMemberInfo;
}

/**
 *  Called by individual thread itself to return itself to the idle list
 */
void ThreadPool::ReturnToIdleList(ThreadMemberInfo *pMemberInfo)
{
   SyncWrap<SyncLock> wrap(pListLock);

   int busyIndex = 0;
   for (int i = 0; i < static_cast<int>(pBusyList->GetSize()); ++i) {
      if (pBusyList->Get(i) == pMemberInfo) {    // find the original index in busy list
         busyIndex = i;
         break;
      }
   }

   for (int i = 0; i < static_cast<int>(pIdleList->GetSize()); ++i) {
      if (pIdleList->Get(i) == NULL) {          // add to empty slot
         pIdleList->Set(i, pMemberInfo);
         break;
      }
   }

   pBusyList->Remove(busyIndex);          // remove reference in busy list
   AtomicDecrementFetch(&numBusyThreads); // update
   AtomicIncrementFetch(&numIdleThreads); // update

   pIdleSem->Post();     // notify there is one idle member
}

/**
 *  Return total number of threads maintained by this pool
 */
u32_t ThreadPool::GetNumThreads()
{
   return numThreads;
}

/**
 *  Get number of idle threads
 */
u32_t ThreadPool::GetNumIdleThreads()
{
   return AtomicRead(&numIdleThreads);
}

/**
 *  Get number of busy threads
 */
u32_t ThreadPool::GetNumBusyThreads()
{
   return AtomicRead(&numBusyThreads);
}

/**
 *  Start all threads and in waiting state
 */
void ThreadPool::StartAllThreads()
{
   for (int i = 0 ; i < static_cast<int>(numThreads); ++i) {
      // make MemberFuncPtr to have a weak reference to ThreadPool (child to parent reference)
      ThreadMemberInfo *pMemberInfo = new ThreadMemberInfo(
                                          new MemberFuncPtr1T<ThreadPool, void, ThreadMemberInfo*>(
                                               this, &ThreadPool::ReturnToIdleList, true));  
      pIdleList->Set(i, pMemberInfo);
      pIdleSem->Post();
      pMemberInfo->StartThread();
   }
}

/**
 *  Stop all threads
 */
void ThreadPool::StopAllThreads()
{
   const int SLEEPWAIT = 1000;

   while (GetNumIdleThreads() < numThreads)  // wait until all threads return to idle list 
      Thread::Sleep(SLEEPWAIT);

   for (int i = 0; i < static_cast<int>(pIdleList->GetSize()); ++i) {
      pIdleList->Get(i)->StopThread();
   }

}

/**
 *  Execute a user function on an idle thread. This will block until an idle thread is present
 *
 *  @param [in] pUserFunc -- user provided function to be executed
 *  @param [in] pUserFuncArg -- argument to be passed to user provided function
 */
void ThreadPool::ExecuteOnIdleThread(ThreadPool::UserFunc *pUserFunc, ptrany_t pUserFuncArg)
{
   ThreadMemberInfo *pMemberInfo = GetIdleMember();
   pMemberInfo->ExecuteJob(pUserFunc, pUserFuncArg);
}

} // end of namespace threading


} // end of namespace cppflib

